using System;
using System.Collections;
using System.Collections.Generic;
using System.IO;
using System.Net;
using System.Net.Sockets;
using System.Security.Cryptography;
using System.Text;
using System.Threading;

using MessageBuffer;

namespace Calibur
{
	public class Sender
	{
		private const int FIBONACCI_NUM = 64;
		private const int CRYPTO_KEY_LEN = 24;
		private const int SHA_LEN = 20;
		private const UInt64 MASK = 0x00000000000000ff;
		private Session session;
		private Queue<Message> queue;
		private MemoryStream buf;
		private TripleDESCryptoServiceProvider des3Provider;
		private SHA1CryptoServiceProvider sha1Provider;
		private byte[] cryptoKey;
		private UInt64 sequence;
		private byte[][] salt;
		private byte[] bytes8;

		public UInt64 Sequence {
			get { return sequence; }
		}

		private UInt64 fibonacci(int n)
		{
			double square5 = Math.Pow(5, 0.5);
			double f1 = 1 / square5;
			double f2 = Math.Pow((1 + square5) / 2, n);
			double f3 = Math.Pow((1 - square5) / 2, n);
			double result = f1 * (f2 - f3);
			return (UInt64) result;
		}
		
		private void InitSalt()
		{
			salt = new byte[FIBONACCI_NUM][];
			UInt64 v;
			for (int i = 0; i < FIBONACCI_NUM; ++i) {
				v = fibonacci(i + 2);
				for (int j = 0; j < sizeof(UInt64); ++j) {
					bytes8 [j] = (byte)((v >> (j << 3)) & MASK);
				}
				salt [i] = sha1Provider.ComputeHash(bytes8);
			}
		}

		public Sender(Session session)
		{
			this.session = session;
			this.queue = new Queue<Message>();
			this.buf = new MemoryStream();
			/* 应用指令报文顺序号，！！！注意不包含心跳等控制指令 */
			this.sequence = 0;
			
			/* 初始化3DES加密provider */
			des3Provider = new TripleDESCryptoServiceProvider();
			des3Provider.Mode = CipherMode.ECB;
			des3Provider.Padding = PaddingMode.PKCS7;
			
			sha1Provider = new SHA1CryptoServiceProvider();
			cryptoKey = new byte[CRYPTO_KEY_LEN];
			
			/* 初始化斐波那契SHA */
			bytes8 = new byte[sizeof(Int64)];
			InitSalt();
		}
		
		private byte[] getShaApplication(UInt64 cts, UInt64 sts, UInt64 seq)
		{
			int i, bits;
			byte b1, b2, b3;
			for (i = 0; i < sizeof(UInt64); ++i) {
				bits = i << 3;
				b1 = (byte)((cts >> bits) & MASK);
				b2 = (byte)((sts >> bits) & MASK);
				b3 = (byte)((seq >> bits) & MASK);
				bytes8 [i] = (byte)(b1 ^ b2 ^ b3);
			}
			
			return sha1Provider.ComputeHash(bytes8);
		}
		
		public void CreateCryptoKey(UInt64 cts, UInt64 sts, UInt64 sequ, byte[] key)
		{
			byte[] shaFibonacci = salt [sequ % (UInt64)FIBONACCI_NUM];
			byte[] shaApplication = getShaApplication(cts, sts, sequ);

			Array.Clear(key, 0, CRYPTO_KEY_LEN);
			Array.Copy(shaFibonacci, key, SHA_LEN);
			for (int i = 0; i < SHA_LEN; ++i) {
				key [i + 4] ^= shaApplication [i];
			}
		}

		public void Send(Message message)
		{
			queue.Enqueue(message);
			lock (this) {
				Monitor.Pulse(this);
			}
		}
		
		public void Reset()
		{
			queue.Clear();
			buf.SetLength(0);
			buf.Seek(0, SeekOrigin.Begin);
			sequence = 0;
		}

		public void Send(MessageBean mbean)
		{
			CommandMessage message = new CommandMessage(mbean, session.MessageBeanFactory);
			Send(message);
		}
		
		private bool SendToPeer(Message message)
		{
			Int32 msgSize = 0;
			buf.SetLength(0);
			
			/* 预留1个int（4个字节）的message size空间 */
			BinaryMessageBeanFormat.PutInt32(buf, msgSize);
			
			/* 1个字节的message type */
			byte messageType = (byte)(int)message.Type;
			if (message.Type == MessageType.App && session.Crypto) {
				messageType |= 0x80;
			}
			BinaryMessageBeanFormat.PutByte(buf, messageType);
			
			/* 消息体，如果messageType==App，则是MessageBean */
			if (message.Type == MessageType.App && session.Crypto) {
				CreateCryptoKey(session.ClientTimestamp, session.ServerTimestamp, sequence, cryptoKey);
				Logger.Trace("RealKey ...");
				Logger.Trace(Logger.ToHexString(cryptoKey));
				CryptoStream cryptoStream = new CryptoStream(buf,
                    des3Provider.CreateEncryptor(cryptoKey, null),
                    CryptoStreamMode.Write);
				message.Serialize(cryptoStream);
				cryptoStream.FlushFinalBlock();
			} else {
				message.Serialize(buf);
			}
			
			/* 回填message size */
			int total = (int)buf.Length;
			msgSize = total - 4;
			buf.Seek(0, SeekOrigin.Begin);
			BinaryMessageBeanFormat.PutInt32(buf, msgSize);
			
			/* 发送数据 */
			int remaining = total;
			int sent = 0;
			int cnt = 0;
			byte[] bytes = buf.GetBuffer();
			while (remaining > 0) {
				cnt = session.Socket.Send(bytes, sent, remaining, SocketFlags.None);
				if (cnt <= 0) {
					return false;
				} else {
					remaining -= cnt;
					sent += cnt;
				}
			}
			
			return true;
		}

		public void ThreadRun()
		{
			Message message = null;
			while (true) {
				lock (this) {
					try {
						Monitor.Wait(this);
					} catch (Exception e) {
						session.onSenderException(e);
						return;
					}
				}

				while (queue.Count > 0) {
					try {
						message = queue.Dequeue();
						if (!SendToPeer(message)) {
							session.onSenderException(new SessionLostException());
							return;
						}
						
						if (message.Type == MessageType.App) {
							++sequence;
						}
					} catch (Exception e) {
						session.onSenderException(e);
						return;
					} finally {
						if (message.Type == MessageType.App) {
							MessageBean mbean = ((CommandMessage)message).MessageBean;
							if (mbean != null) {
								mbean.Release();
							}
						}
					}
				}
			}
		}
	}
	
	internal class MessageReader
	{
		enum RecvPhase
		{
			MessageSize,
			MessageBody
		}
		
		Socket socket;
		int remaining, got, toRead;
		RecvPhase phase;
		byte[] bytes;
		MemoryStream recvBuffer;
			
		public MessageReader(int remaining)
		{
			bytes = new byte[1024];
			recvBuffer = new MemoryStream();
			Reset(RecvPhase.MessageSize, remaining);
		}
			
		public MessageReader() : this(0)
		{
		}
		
		public void Reset(Socket socket)
		{
			this.socket = socket;
			Reset(RecvPhase.MessageSize, 4);
		}
			
		private void Reset(RecvPhase phase, int remaining)
		{
			this.phase = phase;
			this.remaining = remaining;
			this.got = 0;
			this.toRead = 0;
			
			recvBuffer.SetLength(0);
			recvBuffer.Seek(0, SeekOrigin.Begin);
		}
		
		private int socketRecv(byte[] buffer, int start, int count)
		{
			int recvBytes = socket.Receive(buffer, start, count, SocketFlags.None);
			if (recvBytes <= 0) {
				throw new SessionLostException("Broken pipe.");
			} else {
				return recvBytes;
			}
		}
		
		public Message read(MessageBeanFactory factory)
		{
			int msgSize = 0;
			int msgType = 0;
			int cnt = 0;
			
			if (phase == RecvPhase.MessageSize) {
				if (remaining > 0) {
					cnt = socketRecv(bytes, got, remaining);
					remaining -= cnt;
					got += cnt;
				}

				if (remaining == 0) {
					msgSize = BinaryMessageBeanFormat.GetInt32(bytes, 0);
					Reset(RecvPhase.MessageBody, msgSize);
				}
			} else {
				toRead = Math.Min(remaining, bytes.Length);
				cnt = socketRecv(bytes, 0, toRead);
				recvBuffer.Write(bytes, 0, cnt);
				remaining -= cnt;
						
				if (remaining == 0) {
					recvBuffer.Seek(0, SeekOrigin.Begin);
					msgType = recvBuffer.ReadByte();
					Message message = Message.CreateMessage(msgType, factory);
					message.Deserialize(recvBuffer);
					Reset(RecvPhase.MessageSize, 4);
					return message;
				}
			}
			
			return null;
		}
	}

	public class Receiver
	{
		private Session session;
		private MessageReader reader;
		
		public Receiver(Session session)
		{
			this.session = session;
			this.reader = new MessageReader();
		}
		
		public void Reset()
		{
			reader.Reset(session.Socket);
		}

		public void ThreadRun()
		{
			Socket socket = session.Socket;
			reader.Reset(socket);
			Message message = null;
			bool readable;
			int timeout = session.HeartbeatInterval * 1000000;

			while (true) {
				try {
					readable = socket.Poll(timeout, SelectMode.SelectRead);
					if (readable) {
						message = reader.read(session.MessageBeanFactory);
						if (message != null) {
							session.OnMessage(message);
						}
					} else {
						session.Idle();
					}
				} catch (Exception e) {
					session.onReceiverException(e);
					return;
				}
			}
		}
	}

	public class Session
	{
		enum State
		{
			Connecting,		/* 正在建立TCP连接 */
			Requesting,		/* TCP连接已建立，正在申请创建会话 */
			Established,	/* 会话已建立，可以正常收发数据 */
			Closing,		/* 正在关闭连接 */
			Closed			/* 连接已关闭 */
		}

		public enum Error
		{
			SessionNotClosed		= -1,
			ErrorAddressFormat		= -2,
			ConnectRefused			= -3,
			ConnectionLost			= -4
		}
		
		const int DEFAULT_HEARTBEAT_INTERVAL = 60;
		const int DEFAULT_CONNECT_TIMEOUT = 10;

//		private static Regex ipv4Regex = new Regex("[0-9]+(\\.[0-9]+){3}");
		private State currentState;
		private EndPoint hostEP;
		private Socket socket;
		private UInt64 clientTimestamp, serverTimestamp;
		private string sessionKey;
		private int heartbeatInterval;
		private Sender sender;
		private Receiver receiver;
		private Thread senderThread, receiverThread;
		private Queue<MessageBean> queue;
		private bool checkingHeartbeat;
		private HeartbeatMessage hbREQ, hbACK;
		private MessageBeanFactory factory;
		private bool crypto;
		
		public Socket Socket {
			get { return socket; }
		}
		
		public int HeartbeatInterval {
			get { return heartbeatInterval; }
		}
		
		public MessageBeanFactory MessageBeanFactory {
			get { return factory; }
		}
		
		public bool Established {
			get { return currentState == State.Established; }
		}
		
		public UInt64 ClientTimestamp {
			get { return clientTimestamp; }
		}
		
		public UInt64 ServerTimestamp {
			get { return serverTimestamp; }
		}
		
		public bool Crypto {
			get { return crypto; }
		}
		
		public String SessionKey {
			get { return sessionKey; }
		}
		
		public Session(MessageBeanFactory factory) : this(factory, true)
		{
		}

		public Session(MessageBeanFactory factory, bool crypto)
		{
			this.factory = factory;
			this.crypto = crypto;
			
			currentState = State.Closed;
			sender = new Sender(this);
			receiver = new Receiver(this);
			queue = new Queue<MessageBean>();
			socket = null;
			checkingHeartbeat = false;
			hbREQ = new HeartbeatMessage(MessageType.ClientHeartbeat);
			hbACK = new HeartbeatMessage(MessageType.ServerHeartbeatAck);
		}
			
		public void onSenderException(Exception e)
		{
			currentState = State.Closing;
			Logger.Trace("Sender exception:" + e);
		}
		
		public void onReceiverException(Exception e)
		{
			currentState = State.Closing;
			Logger.Trace("Receiver exception:" + e);
		}

		public void Close()
		{
			lock (this) {
				if (socket != null) {
					socket.Close();
					socket = null;
				}

				if (receiverThread != null) {
					if (receiverThread.IsAlive) {
						receiverThread.Interrupt();
					}
					
					receiverThread.Join();
					receiverThread = null;
				}
				
				if (senderThread != null) {
					if (senderThread.IsAlive) {
						senderThread.Interrupt();
					}
					
					senderThread.Join();
					senderThread = null;
				}

				currentState = State.Closed;
			
				queue.Clear();
				checkingHeartbeat = false;
			}
		}
		
		private void StartIOThreads()
		{
			sender.Reset();
			senderThread = new Thread(new ThreadStart(sender.ThreadRun));
			senderThread.Start();
			
			receiver.Reset();
			receiverThread = new Thread(new ThreadStart(receiver.ThreadRun));
			receiverThread.Start();
		}

		public int Open(string host, int port, int heartbeatInterval, int connectingTimeout)
		{
			if (currentState != State.Closed) {
				return (int)Error.SessionNotClosed;
			} else {
				this.heartbeatInterval = heartbeatInterval;
			}
			
			/* 与服务器建立连接 */
			currentState = State.Connecting;
			int errcode = Connect(host, port, connectingTimeout);
			if (errcode != 0) {
				Close();
				return errcode;
			}
			
			/* 把socket设置成为nonblocking模式 */
			socket.Blocking = false;
			/* 启动网络IO线程 */
			StartIOThreads();
			
			/* 向服务器申请建立应用层会话 */
			currentState = State.Requesting;
			errcode = RequestSession(connectingTimeout);
			if (errcode != 0) {
				Close();
				return errcode;
			}

			return 0;
		}
		
		public int Open(string host, int port)
		{
			return Open(host, port, DEFAULT_HEARTBEAT_INTERVAL, DEFAULT_CONNECT_TIMEOUT);
		}

		private int RequestSession(int timeout)
		{
			try {
				clientTimestamp = Message.CurrentTimeMillis();
				RequestSessionMessage rsm = new RequestSessionMessage(clientTimestamp);
				sender.Send(rsm);
			
				for (int i = 0; i < timeout; ++i) {
					Thread.Sleep(1000);
					if (currentState == State.Established) {
						break;
					}
				}
			} catch (Exception e) {
				Logger.Trace(e);
			}

			if (currentState != State.Established) {
				return (int)Error.ConnectionLost;
			}

			return 0;
		}

		private int Connect(string host, int port, int timeout)
		{
			try {
//				IPAddress ipAddress;
//				Match m = ipv4Regex.Match(host);
//				if (m.Success) { /* host 是127.0.0.1这样格式的ip地址 */
//					ipAddress = IPAddress.Parse(host);
//					hostEP = new IPEndPoint(ipAddress, port);
//				} else { /* host是域名 */
//					hostEP = new DnsEndPoint(host, port);
//				}
//				
				IPAddress ipAddress = IPAddress.Parse(host);
				hostEP = new IPEndPoint(ipAddress, port);
			} catch (Exception e) {
				Logger.Trace(e);
				return (int)Error.ErrorAddressFormat;
			}
			
			socket = new Socket(AddressFamily.InterNetwork, SocketType.Stream, ProtocolType.Tcp);
			IAsyncResult result = socket.BeginConnect(hostEP, null, null);
			result.AsyncWaitHandle.WaitOne(timeout * 1000, true);
			return socket.Connected ? 0 : (int)Error.ConnectRefused;

		}
		
		public void Idle()
		{
			if (checkingHeartbeat) { /* 两次检测心跳失败 */
				Logger.Trace("Check heartbeat failed, close session ...");
				Close();
			} else {
				Logger.Trace("Receiver IDLE, checking heartbeat ...");
				checkingHeartbeat = true;
				sender.Send(hbREQ);
			}
		}
		
		public void OnMessage(Message message)
		{
			checkingHeartbeat = false;
			
			switch (message.Type) {
				case MessageType.ClientHeartbeatAck:
					Logger.Trace("Received client heartbeat ACK.");
					break;

				case MessageType.ServerHeartbeat:
					sender.Send(hbACK);
					break;

				case MessageType.App:
					CommandMessage cm = (CommandMessage)message;
					lock (this) {
						queue.Enqueue(cm.MessageBean);
					}
					break;
				
				case MessageType.SessionEstablished:
					SessionEstablishedMessage sem = (SessionEstablishedMessage)message;
					serverTimestamp = sem.EstablishedTimestamp;
					sessionKey = sem.SessionKey;
					currentState = State.Established;
					Logger.Trace("Session established, key={0} ts={1}",
						sessionKey, serverTimestamp);
					break;

				default:
					break;
			}
		}

		public void SendCommand(int commandId, MessageBean mbean)
		{
			if (currentState != State.Established) {
				throw new SessionLostException();
			} else if (senderThread == null) {
				throw new SessionLostException();
			} else if (!senderThread.IsAlive) {
				throw new SessionLostException();
			}

			MessageBean clone = (MessageBean) mbean.Clone();
			clone.commandId = commandId;
			sender.Send(new CommandMessage(clone, factory));
		}

		public MessageBean RecvCommand()
		{
			if (currentState != State.Established) {
				throw new SessionLostException();
			} else if (receiverThread == null) {
				throw new SessionLostException();
			} else if (!receiverThread.IsAlive) {
				throw new SessionLostException();
			}
			
			lock (this) {
				return queue.Count > 0 ? queue.Dequeue() : null;
			}
		}
	}
	
	public class SessionLostException : ApplicationException
	{
		public SessionLostException(string msg) : base(msg)
		{
		}
		
		public SessionLostException() : base()
		{
		}
	}
}
